package com.ly.demo.scheduled;

import cn.hutool.json.JSONUtil;
import com.ly.demo.entity.KafkaVo;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import javax.annotation.Resource;

import static com.ly.demo.utils.Constant.BAAS_CONTACT_USER_NOTICE;

/**
 * @Author liuyang
 * @Date 2024/3/15 17:49
 **/
@Slf4j
@Service
public class KafkaTemplateTask {
    @Resource
    private KafkaTemplate<String, Object> kafkaTemplate;

    /**
     * 通讯录人员相关 kafka 消息发送
     *
     * @param kafkaVo
     */
    public void sendUserNoticeJustData(KafkaVo<?> kafkaVo) {
        try {
            ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(BAAS_CONTACT_USER_NOTICE, JSONUtil.toJsonStr(kafkaVo.getData()));
            future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
                @Override
                public void onSuccess(SendResult<String, Object> result) {
                    log.info("[KafkaMsgSend]-[success] topic:{}, params: {}, result {}", BAAS_CONTACT_USER_NOTICE, kafkaVo,
                            JSONUtil.toJsonStr(result));
                }

                @Override
                public void onFailure(Throwable ex) {
                    log.info("[KafkaMsgSend]-[failed] topic:" + BAAS_CONTACT_USER_NOTICE + ", params:" + kafkaVo, ex);
                }
            });
            log.info("sendSaveUserNotice sendKafka success - dto:{}", kafkaVo);
        } catch (Exception e) {
            log.error("sendSaveUserNotice sendKafka fail - dto:{}", kafkaVo);
        }
    }
}
